package com.hexinfo.dmpro.sparing.service.impl;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import com.hexinfo.dmpro.sparing.dto.TableDistCPDTO;
import com.hexinfo.dmpro.sparing.service.AsyncDropDataService;
import com.hexinfo.dmpro.sparing.service.BackupSourceService;
import com.hexinfo.dmpro.sparing.service.DataRefreshService;
import com.hexinfo.dmpro.sparing.util.Logger;
import com.xqfunds.job.core.log.XxlJobLogger;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.Statement;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class AsyncDropDataServiceImpl implements AsyncDropDataService {

    @Autowired
    @Qualifier("connPoolHiveTarget")
    private DataSource connPoolHiveTarget;

    //表完成数量
    private Map<String, Integer> tableSizeMap = new ConcurrentHashMap<String, Integer>();
    private synchronized void addTableSize(String souName) {
        tableSizeMap.put(souName, tableSizeMap.getOrDefault(souName,0)+1);
    }
    public synchronized Integer getTableSize(String souName) {
        return tableSizeMap.get(souName);
    }
    public synchronized void deleteTableSize(String souName) {
        tableSizeMap.remove(souName);
    }

    //结束标志
    private Map<String, Boolean> endFlagMap = new ConcurrentHashMap<String, Boolean>();
    public synchronized void updateEndFlag(String souName,Boolean state) {
        endFlagMap.put(souName, state);
    }
    public synchronized Boolean getEndFlag(String souName) {
        return endFlagMap.get(souName);
    }
    public synchronized void deleteEndFlag(String souName) {
        endFlagMap.remove(souName);
    }

    //记录任务执行是否有异常
    private Map<String, Boolean> taskStateMap = new ConcurrentHashMap<String, Boolean>();
    public synchronized void updateTaskState(String souName,Boolean state) {
        taskStateMap.put(souName, state);
    }
    public synchronized Boolean getTaskState(String souName) {
        return taskStateMap.get(souName);
    }
    public synchronized void deleteTaskState(String souName) {
        taskStateMap.remove(souName);
    }

    @Override
    @Async("dropTaskExecutor")
    public void execShell(String dropSql,String souName,int sizes,FileAppender heatlog) {
        //执行dropSql
        String str = execDropData(dropSql, souName, heatlog);
        addTableSize(souName);    //表完成数量加一
        Integer integer = getTableSize(souName);
        logLog(heatlog,"（"+integer+str,"1");
//        Logger.log("（"+integer+str);
        logLog(heatlog,"","2"); //刷进日志中
        if (integer.equals(sizes) || integer == sizes){
            deleteTableSize(souName);  //移除key
            logLog(heatlog,"【源全部表dropSQL执行完成】","1"); //刷进日志中
//            Logger.log("【源全部表dropSQL执行完成】");
            logLog(heatlog,"","2"); //刷进日志中
            updateEndFlag(souName,true);
        }
    }

    /**
     * 执行drop语句
     */
    private String execDropData(String dropSql,String souName,FileAppender heatlog) {
        try {
            Connection connection = connPoolHiveTarget.getConnection();
            try{
                @Cleanup
                Statement st = connection.createStatement();
                int i = st.executeUpdate(dropSql);
                return "）执行dropSql："+dropSql+"--执行状态："+i;
            }catch (Exception e){
                log.error("执行dropSql："+dropSql+"【异常】"+e);
                updateTaskState(souName,false);    //修改任务异常状态
                return "）执行dropSql："+dropSql+"【异常】"+e;
            }finally {
                try {
                    if (connection != null) {
                        connection.close();
                    }
                } catch (Exception e) {
                    log.error("关闭连接异常"+e);
                    logLog(heatlog,"关闭连接异常："+e,"1");
//                    Logger.log("关闭连接异常："+e);
                    updateTaskState(souName,false);    //修改任务异常状态
                }
            }
        }catch (Exception e){
            log.error("执行dropSql："+dropSql+"【获取连接池连接异常】："+e);
            updateTaskState(souName,false);    //修改任务异常状态
            return "）执行dropSql："+dropSql+"【获取连接池连接异常】："+e;
        }
    }

    /**
     * 加锁，防止并发问题
     */
    private synchronized void logLog(FileAppender heatlog, String logStr ,String str) {
        if (str.equals("1")){
            heatlog.append(logTime(logStr));
        }else {
            heatlog.flush();     //刷进日志中
        }
    }

    /**
     * 日志加日期时间
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
